package com.baitiaojun.moments.handler;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.baitiaojun.api.domain.moments.UserDailyMoments;
import com.baitiaojun.api.domain.moments.UserMoments;
import com.baitiaojun.api.domain.moments.UserVideoMoments;
import com.baitiaojun.api.domain.user.UserFollowing;
import com.baitiaojun.api.util.ConstantUtil;
import com.baitiaojun.api.util.SnowFlakeUtil;
import com.baitiaojun.moments.config.RedissonConfig;
import com.baitiaojun.moments.config.properties.MqUserMomentsProperties;
import com.baitiaojun.moments.feign.UserFeign;
import com.baitiaojun.moments.service.DailyMomentsService;
import com.baitiaojun.moments.service.VideoMomentsService;
import com.baitiaojun.moments.support.RedisSupport;
import com.rabbitmq.client.Channel;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.ObjectUtils;
import org.redisson.api.RBloomFilter;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;
import org.springframework.retry.annotation.Backoff;
import org.springframework.retry.annotation.Retryable;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * @类名 RabbitMqSupport
 * @描述 TODO
 * @作者 白条君
 * @创建日期 2023/1/16 15:12
 * @版本 1.0
 */
@Slf4j
@Component
public class RabbitMqHandler {

    @Resource
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private MqUserMomentsProperties properties;

    @Resource
    private RedisSupport redisSupport;

    @Autowired
    private UserFeign userFeign;

    @Lazy
    @Autowired
    private VideoMomentsService videoMomentsService;

    @Lazy
    @Autowired
    private DailyMomentsService dailyMomentsService;

    @Autowired
    private RedissonConfig redisson;

    @Retryable(maxAttempts = 3, backoff = @Backoff(delay = 5 * 1000, multiplier = 3))
    @Async
    public void publishMoments(UserMoments userMoments, Long id) {
        log.info("发送消息");
        CorrelationData correlationData = new CorrelationData();
        correlationData.setId(String.valueOf(id));
        Message message = MessageBuilder.withBody(JSONObject.toJSONString(userMoments).getBytes(StandardCharsets.UTF_8)).build();
        MessageProperties messageProperties = message.getMessageProperties();
        messageProperties.setDeliveryMode(MessageDeliveryMode.PERSISTENT);
        //设置deliveryTag用于
        messageProperties.setCorrelationId(correlationData.getId());
        //设置一个全局messageId用于幂等性操作
        messageProperties.setMessageId(String.valueOf(SnowFlakeUtil.getNextId()));
        rabbitTemplate.convertAndSend(properties.getExchange(), properties.getRoutingKey(), message, correlationData);
    }

    @Async
    @SneakyThrows
    @RabbitListener(queues = {"mirror_moments_queue", "mirror_backup_moments_queue"}, containerFactory = "listenerContainerFactory")
    public void receiveMoments(Message message, Channel channel) {
        MessageProperties properties = message.getMessageProperties();
        /**
         * consumer在给Broker返回ack途中网络中断，Broker未收到确认信息，根据RabbitMQ的重试补偿机制，
         * 则会把这条消息再重发给其他的消费者或等网络重连后再发送给该消费者，造成了消息的重复消费
         * 通过redis的分布式死锁保证消费的幂等性,利用setnx不可重复设值的特点，返回值为true说明没有被消费过，返回值为false说明已经被消费过了
         * 注意：broker重发这个消息的时候会携带最开始设置的同一个getMessageId
         */
        long deliveryTag = properties.getDeliveryTag();
        String msg = new String(message.getBody());
        String key = properties.getMessageId();
        try {
            //幂等性操作，防止重复消费
            if (redisSupport.setNx(key, null, 1L, TimeUnit.MINUTES)) {
                //删除redis原来记录
                this.deleteRedis(msg);
                redisSupport.delete(key);
            }
            //手动应答
            channel.basicAck(deliveryTag, false);
            log.info("收到消息：{}", msg);
        } catch (Exception e) {
            //接收消费异常就删除setnx，然后拒绝接受消息并进入死信队列
            redisSupport.delete(key);
            channel.basicNack(deliveryTag, false, false);
            log.error(e.getMessage());
        }
    }

    @Async
    @RabbitListener(queues = "mirror_warning_moments_queue")
    public void receiveWarning(Message message) {
        log.warn("警告！路由不可达，交由备份交换机和备份队列处理，消息：{}", new String(message.getBody()));
    }

    @Async
    @SneakyThrows
    @RabbitListener(queues = "mirror_dead_moments_queue", containerFactory = "listenerContainerFactory")
    public void receiveDead(Message message, Channel channel) {
        MessageProperties properties = message.getMessageProperties();
        long deliveryTag = properties.getDeliveryTag();
        String msg = new String(message.getBody());
        String key = properties.getMessageId();
        try {
            //幂等性操作，防止重复消费
            if (redisSupport.setNx(key, null, 1L, TimeUnit.MINUTES)) {
                //删除redis原来记录
                this.deleteRedis(msg);
                redisSupport.delete(key);
            }
            //手动应答
            channel.basicAck(deliveryTag, false);
            log.info("收到消息：{}", msg);
        } catch (Exception e) {
            log.error("死信队列消费失败");
            //短信发送提示是哪条消息出现问题，进行人工处理

        }
    }

    private void deleteRedis(String msg) throws Exception {
        UserMoments userMoments = JSONObject.toJavaObject(JSONObject.parseObject(msg), UserMoments.class);
        Long userId = userMoments.getUserid();
        //删除粉丝的记录
        redisSupport.asyncDelete(userId);
        //布隆过滤器添加新值
        RBloomFilter<Object> bloomFilter = redisson.bloomFilter();
        bloomFilter.add("subscribed" + userId);
        bloomFilter.add("self" + userId);
    }



    //弃用
    @Deprecated
    private void broadcast(String msg) throws Exception {
        UserMoments userMoments = JSONObject.toJavaObject(JSONObject.parseObject(msg), UserMoments.class);
        Long userid = userMoments.getUserid();
        Long contentId = userMoments.getContentid();
        //0：视频，1：日常,2：专栏
        String type = userMoments.getType();
        if (type.equals(ConstantUtil.VIDEO_TYPE)) {
            UserVideoMoments videoMoments = videoMomentsService.getUserVideoMoments(contentId, userid);
            this.broadcast(videoMoments, userid, contentId);
        } else if (type.equals(ConstantUtil.DAILY_TYPE)) {
            UserDailyMoments userDailyMoments = dailyMomentsService.getDailyMoments(contentId, userid);
            this.broadcast(userDailyMoments, userid, contentId);
        } else if (type.equals(ConstantUtil.SPECIAL_COLUMN_TYPE)) {

        }
    }

    //弃用
    @Deprecated
    private void broadcast(Object userMoments, Long userId, Long contentId) throws Exception {
        //发布给粉丝
        Object data = userFeign.getFans(userId).get("data");
        if (ObjectUtils.isNotEmpty(data)) {
            List<UserFollowing> userFollowings = JSONArray.parseArray(JSONObject.toJSONString(data), UserFollowing.class);
            for (UserFollowing userFollowing : userFollowings) {
                redisSupport.syncRedis(userFollowing.getUserid(), userMoments, contentId);
            }
        }
    }
}